package com.li.flink.datastream

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object TransformationOp {
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    val inputPath = "/Users/lijiacen/Downloads/test.txt"
    environment.setParallelism(1)
    val textStream = environment.readTextFile(inputPath)
    textStream.map(m => {
      m
    }).print() //map算子作用到DataStream上，作用到每一个元素，并产生新的DataStream
    textStream.flatMap(_.split(",")).print()
    textStream.flatMap(_.split(",")).filter(m => {
      val num = m.toInt
      num > 2
    }).print() //map算子作用到DataStream上，作用到每一个元素，并返回结果为true的元素，并产生新的DataStream
    textStream.flatMap(_.split(",")).map((_, 1)).keyBy(_._1).sum(1).print()
    textStream.flatMap(_.split(",")).map(m => {
      (m.toInt, 1)
    }).keyBy(_._1).reduce((x, y) => {
      (x._1, x._2 + y._2)
    }).print()
    environment.execute("TransformationOp")
  }
}
